package com.navinfo.platform.trip.analysis.flink.config;

import com.navinfo.platform.trip.common.config.LoadConf;
import org.apache.flink.api.java.utils.ParameterTool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * 算法链的配置
 * @author: web
 * @data: 2020-04-08
 **/
public class AlgorithmConfig {
    private static final Logger logger = LoggerFactory.getLogger(AlgorithmConfig.class);

    /**
     * 加载算法链的配置信息<br/>
     * 并使用命令行中传入的参数覆盖配置文件中的值<br/>
     * 命令行通过--key或-key value传值
     * @param args 命令行参数
     * @return  算法配置信息
     */
    @SuppressWarnings("unchecked")
    public static ParameterTool load(String[] args)  {
        Map algConfig = new HashMap<>(LoadConf.LoadYaml("config_common.yaml"));
        algConfig.putAll(LoadConf.LoadYaml("config_threshold.yaml"));

        //将value转换成String类型
        algConfig.forEach((key,value) -> {
            algConfig.put(key, value.toString());
        });

        ParameterTool fileParameter = ParameterTool.fromMap(algConfig);
        ParameterTool argsParameter = ParameterTool.fromArgs(args);

        //命令行中的配置覆盖file中的配置
        ParameterTool parameterTool = fileParameter.mergeWith(argsParameter);

        logger.info("加载算法链的配置开始......");
        parameterTool.toMap().forEach((s, s2) -> logger.info(">>>{}----{}", s, s2));
        logger.info("加载算法链的配置结束......");
        return parameterTool;
    }
}
